package com.chb.flink.source

import java.util.Properties

import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.kafka.{FlinkKafkaConsumer, KafkaDeserializationSchema}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer

object KafkaSourceByKeyValue {
    def main(args: Array[String]): Unit = {
        //初始化Flink的Streaming（流计算）上下文执行环境
        val streamEnv = StreamExecutionEnvironment.getExecutionEnvironment
        streamEnv.setParallelism(1)
        //导入隐式转换
        import org.apache.flink.streaming.api.scala._

        val props = new Properties()
        props.setProperty("bootstrap.servers", "ShServer:9092")
        props.setProperty("group.id", "fink02")
        props.setProperty("key.deserializer", classOf[StringDeserializer].getName)
        props.setProperty("value.deserializer", classOf[StringDeserializer].getName)
        props.setProperty("auto.offset.reset", "latest")

        //设置kafka为数据源
        val stream = streamEnv.addSource(new FlinkKafkaConsumer[(String, String)]("test", new KafkaDeserializationSchema[(String, String)] {
            override def isEndOfStream(t: (String, String)): Boolean = false

            override def deserialize(consumerRecord: ConsumerRecord[Array[Byte], Array[Byte]]): (String, String) = {
                if (consumerRecord != null) {
                    var key = "NULL"
                    var value = "NULL"
                    if (consumerRecord.key() != null) {
                        key = new String(consumerRecord.key())
                    }
                    if (consumerRecord.value() != null) {
                        value = new String(consumerRecord.value())
                    }
                    (key, value)

                } else {
                    ("NULL", "NULL")
                }
            }

            // 设置返回类型为二元组
            override def getProducedType: TypeInformation[(String, String)] = createTuple2TypeInformation(createTypeInformation[String],
                createTypeInformation[String])
        }, props))

        stream.print()
        streamEnv.execute()
    }
}